package com.lambkit.core.event;

import cn.hutool.core.util.ReflectUtil;
import cn.hutool.log.Log;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * @author yangyong(孤竹行)
 */
public class AsyncEventBus extends EventBus {
    private static final Log log = Log.get(AsyncEventBus.class);

    private ExecutorService executorService;

    public AsyncEventBus(ExecutorService executorService) {
        executorService = executorService;
    }

    @Override
    public void register(Class<? extends IEventListener> listener) {
        for (IEventListener ml : listeners) {
            if (listener == ml.getClass()) {
                return;
            }
        }

        IEventListener eventListener = ReflectUtil.newInstance(listener);
        if (eventListener == null) {
            return;
        }

        listeners.add(eventListener);
    }

    @Override
    public void post(IEvent event) {
        BlockingQueue<IEventListener> queue = new LinkedBlockingQueue<>();
        for (IEventListener listener : listeners) {
            //listener.onEvent(event);
            queue.add(listener);
        }
        while (!queue.isEmpty()) {
            IEventListener listener = queue.poll();
            listener.onEvent(event);
        }
        for (IEventListener listener : listeners) {
            executorService.execute(new Runnable() {
                @Override
                public void run() {
                    try {
                        //if (JFinal.me().getConstants().getDevMode()) {
                            //Printer.print(this, "event"(String.format("listener[%s]-->>onMessage(%s) in async", listener, message));
                        //}
                        listener.onEvent(event);
                    } catch (Throwable e) {
                        log.error(String.format("listener[%s] onEvent is error! ", listener.getClass()), e);
                    }
                }
            });
        }
    }
}
